package org.dromara.business.service;

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.business.config.TelegramBotProperties;
import org.dromara.business.model.TelegramMessage;
import org.dromara.common.json.utils.JsonUtils;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientException;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 异步Telegram通知服务
 *
 * 功能特性：
 * 1. 异步消息发送，支持高并发
 * 2. 消息队列管理，优先级排序
 * 3. 智能重试机制，指数退避策略
 * 4. 速率限制，符合Telegram API限制
 * 5. 批量处理，提升效率
 * 6. 健康监控，实时状态检查
 * 7. 故障转移，多重保障机制
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class AsyncTelegramNotificationService implements HealthIndicator {

    private final WebClient.Builder webClientBuilder;
    private final TelegramBotProperties telegramBotProperties;
    private final TelegramRateLimiter rateLimiter;

    private static final String TELEGRAM_API_URL = "https://api.telegram.org/bot";

    // 消息队列 - 按优先级排序
    private final PriorityQueue<TelegramMessage> messageQueue = new PriorityQueue<>(
        Comparator.comparing(TelegramMessage::getPriority)
            .thenComparing(TelegramMessage::getCreateTime)
    );

    // 重试队列 - 延迟重试的消息
    private final DelayQueue<DelayedTelegramMessage> retryQueue = new DelayQueue<>();

    // 统计信息
    private final AtomicLong totalSentMessages = new AtomicLong(0);
    private final AtomicLong totalFailedMessages = new AtomicLong(0);
    private final AtomicInteger queueSize = new AtomicInteger(0);
    private final AtomicInteger retryQueueSize = new AtomicInteger(0);

    // 最近发送状态
    private volatile LocalDateTime lastSuccessfulSend = LocalDateTime.now();
    private volatile String lastError = null;
    private volatile LocalDateTime lastErrorTime = null;

    // 线程池配置
    private final ExecutorService messageProcessingExecutor = Executors.newFixedThreadPool(3, r -> {
        Thread t = new Thread(r, "telegram-sender-" + System.currentTimeMillis());
        t.setDaemon(true);
        return t;
    });

    /**
     * 延迟消息包装器
     */
    private static class DelayedTelegramMessage implements Delayed {
        @Getter
        private final TelegramMessage message;
        private final long executeTime;

        public DelayedTelegramMessage(TelegramMessage message, long delayMs) {
            this.message = message;
            this.executeTime = System.currentTimeMillis() + delayMs;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(executeTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            return Long.compare(this.executeTime, ((DelayedTelegramMessage) o).executeTime);
        }

    }

    /**
     * 异步发送消息
     */
    @Async("taskExecutor")
    public CompletableFuture<Boolean> sendMessageAsync(TelegramMessage message) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                // 设置默认聊天ID
                if (message.getChatId() == null) {
                    message.setChatId(telegramBotProperties.getChat().getId());
                }

                // 检查消息是否过期
                if (message.isExpired()) {
                    log.warn("Message expired, discarding: {}", message.getMessageId());
                    return false;
                }

                // 添加到队列
                synchronized (messageQueue) {
                    messageQueue.offer(message);
                    queueSize.incrementAndGet();
                }

                log.debug("Message queued: type={}, priority={}, queue_size={}",
                         message.getType(), message.getPriority(), queueSize.get());

                return true;
            } catch (Exception e) {
                log.error("Failed to queue message", e);
                return false;
            }
        }, messageProcessingExecutor);
    }

    /**
     * 批量发送消息
     */
    @Async("taskExecutor")
    public CompletableFuture<Map<String, Boolean>> sendBatchMessagesAsync(List<TelegramMessage> messages) {
        return CompletableFuture.supplyAsync(() -> {
            Map<String, Boolean> results = new HashMap<>();

            for (TelegramMessage message : messages) {
                try {
                    boolean success = sendMessageAsync(message).get(5, TimeUnit.SECONDS);
                    results.put(message.getMessageId(), success);
                } catch (Exception e) {
                    log.error("Failed to send message in batch: {}", message.getMessageId(), e);
                    results.put(message.getMessageId(), false);
                }
            }

            return results;
        }, messageProcessingExecutor);
    }

    /**
     * 实际发送消息到Telegram API
     */
    @Retryable(value = {WebClientException.class}, maxAttempts = 1, // 不在这里重试，使用自定义重试逻辑
               backoff = @Backoff(delay = 1000, multiplier = 2))
    private boolean sendToTelegramAPI(TelegramMessage message) {
        try {
            // 检查速率限制
            boolean isPrivateChat = !message.getChatId().startsWith("-");
            if (!rateLimiter.allowRequest(message.getChatId(), isPrivateChat)) {
                long waitTime = rateLimiter.getSuggestedWaitTime(message.getChatId(), isPrivateChat);
                log.debug("Rate limit exceeded, scheduling retry in {}ms", waitTime);
                scheduleRetry(message, waitTime);
                return false;
            }

            // 构建请求体
            Map<String, Object> payload = buildMessagePayload(message);

            // 发送请求
            String response = webClientBuilder.build()
                .post()
                .uri(TELEGRAM_API_URL + telegramBotProperties.getBot().getToken() + "/sendMessage")
                .bodyValue(payload)
                .retrieve()
                .bodyToMono(String.class)
                .retryWhen(Retry.backoff(2, Duration.ofSeconds(1))
                    .filter(throwable -> !(throwable instanceof WebClientResponseException.TooManyRequests)))
                .timeout(Duration.ofSeconds(30))
                .onErrorResume(WebClientResponseException.TooManyRequests.class, ex -> {
                    // 处理429错误（Too Many Requests）
                    log.warn("Telegram API rate limit hit: {}", ex.getMessage());
                    long retryAfter = extractRetryAfter(ex);
                    scheduleRetry(message, retryAfter * 1000);
                    return Mono.empty();
                })
                .block();

            if (response != null) {
                // 记录成功发送
                rateLimiter.recordRequest(message.getChatId(), isPrivateChat);
                totalSentMessages.incrementAndGet();
                lastSuccessfulSend = LocalDateTime.now();
                message.setStatus(TelegramMessage.MessageStatus.SUCCESS);

                log.debug("Message sent successfully: type={}, chatId={}",
                         message.getType(), message.getChatId());
                return true;
            } else {
                // 处理空响应
                handleSendFailure(message, "Empty response from Telegram API");
                return false;
            }

        } catch (WebClientResponseException.TooManyRequests ex) {
            // 处理429错误
            long retryAfter = extractRetryAfter(ex);
            log.warn("Rate limited by Telegram API, retry after: {}s", retryAfter);
            scheduleRetry(message, retryAfter * 1000);
            return false;

        } catch (Exception ex) {
            log.error("Failed to send message to Telegram API: type={}, error={}",
                     message.getType(), ex.getMessage(), ex);
            handleSendFailure(message, ex.getMessage());
            return false;
        }
    }

    /**
     * 构建消息载荷
     */
    private Map<String, Object> buildMessagePayload(TelegramMessage message) {
        Map<String, Object> payload = new HashMap<>();
        payload.put("chat_id", message.getChatId());
        payload.put("text", message.getContent());

        if (message.getEnableMarkdown()) {
            payload.put("parse_mode", "Markdown");
        }

        if (message.getDisableWebPagePreview()) {
            payload.put("disable_web_page_preview", true);
        }

        return payload;
    }

    /**
     * 处理发送失败
     */
    private void handleSendFailure(TelegramMessage message, String error) {
        message.incrementRetryCount();
        message.setErrorMessage(error);
        message.setStatus(TelegramMessage.MessageStatus.FAILED);

        totalFailedMessages.incrementAndGet();
        lastError = error;
        lastErrorTime = LocalDateTime.now();

        // 如果可以重试，加入重试队列
        if (message.canRetry() && !message.isExpired()) {
            long delayMs = calculateRetryDelay(message.getRetryCount());
            scheduleRetry(message, delayMs);
            log.info("Scheduling retry for message: type={}, attempt={}/{}, delay={}ms",
                    message.getType(), message.getRetryCount(), message.getMaxRetries(), delayMs);
        } else {
            log.error("Message failed permanently: type={}, retries={}/{}, error={}",
                     message.getType(), message.getRetryCount(), message.getMaxRetries(), error);
        }
    }

    /**
     * 安排重试
     */
    private void scheduleRetry(TelegramMessage message, long delayMs) {
        message.setStatus(TelegramMessage.MessageStatus.PENDING);
        message.setNextExecuteTime(LocalDateTime.now().plusNanos(delayMs * 1_000_000));

        DelayedTelegramMessage delayedMessage = new DelayedTelegramMessage(message, delayMs);
        retryQueue.offer(delayedMessage);
        retryQueueSize.incrementAndGet();
    }

    /**
     * 计算重试延迟（指数退避）
     */
    private long calculateRetryDelay(int retryCount) {
        // 指数退避：1s, 2s, 4s, 8s...
        return Math.min(1000L * (1L << (retryCount - 1)), 30000L);
    }

    /**
     * 从429响应中提取重试延迟时间
     */
    private long extractRetryAfter(WebClientResponseException.TooManyRequests ex) {
        try {
            String body = ex.getResponseBodyAsString();
            Map<String, Object> response = JsonUtils.parseObject(body, Map.class);
            Object retryAfter = response.get("parameters");
            if (retryAfter instanceof Map) {
                Object retryAfterValue = ((Map<?, ?>) retryAfter).get("retry_after");
                if (retryAfterValue instanceof Number) {
                    return ((Number) retryAfterValue).longValue();
                }
            }
        } catch (Exception e) {
            log.debug("Failed to parse retry_after from response", e);
        }
        return 60; // 默认60秒
    }

    /**
     * 定期处理消息队列
     */
    @Scheduled(fixedDelay = 1000) // 每秒执行一次
    public void processMessageQueue() {
        try {
            // 处理主队列
            processMainQueue();

            // 处理重试队列
            processRetryQueue();

            // 清理速率限制器
            if (System.currentTimeMillis() % 60000 < 1000) {
                rateLimiter.cleanupExpiredCounters();
            }

        } catch (Exception e) {
            log.error("Error processing message queue", e);
        }
    }

    /**
     * 处理主消息队列
     */
    private void processMainQueue() {
        int processed = 0;
        int maxBatchSize = 10; // 每次最多处理10条消息

        synchronized (messageQueue) {
            while (!messageQueue.isEmpty() && processed < maxBatchSize) {
                TelegramMessage message = messageQueue.poll();
                if (message != null) {
                    queueSize.decrementAndGet();

                    // 异步处理消息
                    CompletableFuture.runAsync(() -> {
                        message.setStatus(TelegramMessage.MessageStatus.SENDING);
                        sendToTelegramAPI(message);
                    }, messageProcessingExecutor);

                    processed++;
                }
            }
        }

        if (processed > 0) {
            log.debug("Processed {} messages from main queue, remaining: {}", processed, queueSize.get());
        }
    }

    /**
     * 处理重试队列
     */
    private void processRetryQueue() {
        int processed = 0;
        int maxRetries = 5; // 每次最多处理5条重试消息

        while (processed < maxRetries) {
            DelayedTelegramMessage delayedMessage = retryQueue.poll();
            if (delayedMessage == null) {
                break; // 队列为空或没有到期的消息
            }

            retryQueueSize.decrementAndGet();
            TelegramMessage message = delayedMessage.getMessage();

            // 检查消息是否过期
            if (message.isExpired()) {
                log.warn("Discarding expired retry message: type={}", message.getType());
                continue;
            }

            // 异步处理重试消息
            CompletableFuture.runAsync(() -> {
                message.setStatus(TelegramMessage.MessageStatus.SENDING);
                sendToTelegramAPI(message);
            }, messageProcessingExecutor);

            processed++;
        }

        if (processed > 0) {
            log.debug("Processed {} retry messages, remaining: {}", processed, retryQueueSize.get());
        }
    }

    /**
     * 获取队列统计信息
     */
    public Map<String, Object> getQueueStatistics() {
        Map<String, Object> stats = new HashMap<>();
        stats.put("mainQueueSize", queueSize.get());
        stats.put("retryQueueSize", retryQueueSize.get());
        stats.put("totalSent", totalSentMessages.get());
        stats.put("totalFailed", totalFailedMessages.get());
        stats.put("lastSuccessfulSend", lastSuccessfulSend);
        stats.put("lastError", lastError);
        stats.put("lastErrorTime", lastErrorTime);
        stats.put("rateLimitStatus", rateLimiter.getRateLimitStatus());
        return stats;
    }

    /**
     * 健康检查
     */
    @Override
    public Health health() {
        try {
            Health.Builder healthBuilder = Health.up();

            // 检查队列大小
            int mainQueue = queueSize.get();
            int retryQueue = retryQueueSize.get();

            if (mainQueue > 1000 || retryQueue > 100) {
                healthBuilder = Health.down();
                healthBuilder.withDetail("issue", "Queue size too large");
            }

            // 检查最近的发送状态
            if (lastSuccessfulSend.isBefore(LocalDateTime.now().minusMinutes(5))) {
                healthBuilder.withDetail("warning", "No successful sends in last 5 minutes");
            }

            // 添加统计信息
            healthBuilder.withDetail("mainQueueSize", mainQueue);
            healthBuilder.withDetail("retryQueueSize", retryQueue);
            healthBuilder.withDetail("totalSent", totalSentMessages.get());
            healthBuilder.withDetail("totalFailed", totalFailedMessages.get());
            healthBuilder.withDetail("successRate", calculateSuccessRate());
            healthBuilder.withDetail("lastSuccessfulSend", lastSuccessfulSend);

            if (lastError != null) {
                healthBuilder.withDetail("lastError", lastError);
                healthBuilder.withDetail("lastErrorTime", lastErrorTime);
            }

            return healthBuilder.build();

        } catch (Exception e) {
            return Health.down()
                .withDetail("error", e.getMessage())
                .build();
        }
    }

    /**
     * 计算成功率
     */
    private double calculateSuccessRate() {
        long total = totalSentMessages.get() + totalFailedMessages.get();
        if (total == 0) return 100.0;
        return (double) totalSentMessages.get() / total * 100.0;
    }

    /**
     * 清理资源
     */
    public void shutdown() {
        log.info("Shutting down AsyncTelegramNotificationService...");

        // 等待队列处理完成
        int maxWaitSeconds = 30;
        int waitCount = 0;
        while ((queueSize.get() > 0 || retryQueueSize.get() > 0) && waitCount < maxWaitSeconds) {
            try {
                Thread.sleep(1000);
                waitCount++;
                log.info("Waiting for queues to empty... main: {}, retry: {}",
                        queueSize.get(), retryQueueSize.get());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }

        // 关闭线程池
        messageProcessingExecutor.shutdown();
        try {
            if (!messageProcessingExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
                messageProcessingExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            messageProcessingExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }

        log.info("AsyncTelegramNotificationService shutdown completed");
    }
}
